#pragma once

#include <rocksdb/slice.h>

#include <memory>
#include <string>

#include "slash/coding.h"

class BaseDataKey {
 public:
  BaseDataKey(const rocksdb::Slice &key, int32_t version,
              const rocksdb::Slice &data)
      : start_(nullptr), key_(key), version_(version), data_(data) {}

  ~BaseDataKey() {
    if (start_ != space_) {
      delete[] start_;
    }
  }

  const rocksdb::Slice Encode() {
    size_t usize = key_.size() + data_.size();
    size_t needed = usize + sizeof(int32_t) * 2;
    char *dst_ptr;
    if (needed <= sizeof(space_)) {
      dst_ptr = space_;
    } else {
      dst_ptr = new char[needed];

      if (start_ != space_) {
        delete[] start_;
      }
    }

    start_ = dst_ptr;
    slash::EncodeFixed32(dst_ptr, key_.size());
    dst_ptr += sizeof(int32_t);
    ::memcpy(dst_ptr, key_.data(), key_.size());
    dst_ptr += key_.size();
    slash::EncodeFixed32(dst_ptr, version_);
    dst_ptr += sizeof(int32_t);
    memcpy(dst_ptr, data_.data(), data_.size());
    return rocksdb::Slice(start_, needed);
  }

 protected:
  char space_[256];
  char *start_;
  rocksdb::Slice key_;
  int32_t version_;
  rocksdb::Slice data_;
};

class ParsedBaseDataKey {
 public:
  ParsedBaseDataKey(const std::string *key) {
    const char *data_ptr = key->data();
    int32_t key_len = slash::DecodeFixed32(data_ptr);
    data_ptr += sizeof(int32_t);
    key_ = rocksdb::Slice(data_ptr, key_len);
    data_ptr += key_len;
    version_ = slash::DecodeFixed32(data_ptr);
    data_ptr += sizeof(int32_t);
    data_ =
        rocksdb::Slice(data_ptr, key->size() - key_len + 2 * sizeof(int32_t));
  }

  ParsedBaseDataKey(const rocksdb::Slice &key) {
    const char *data_ptr = key.data();
    int32_t key_len = slash::DecodeFixed32(data_ptr);
    data_ptr += sizeof(int32_t);
    key_ = rocksdb::Slice(data_ptr, key_len);
    data_ptr += key_len;
    version_ = slash::DecodeFixed32(data_ptr);
    data_ptr += sizeof(int32_t);
    data_ =
        rocksdb::Slice(data_ptr, key.size() - key_len + 2 * sizeof(int32_t));
  }

 protected:
  rocksdb::Slice key_;
  int32_t version_;
  rocksdb::Slice data_;
};

class ParsedHashesDataKey : public ParsedBaseDataKey {
 public:
  explicit ParsedHashesDataKey(const std::string *key)
      : ParsedBaseDataKey(key) {}
  explicit ParsedHashesDataKey(const rocksdb::Slice &key)
      : ParsedBaseDataKey(key) {}
  rocksdb::Slice field() { return data_; }
};

class ParsedSetsMemberKey : public ParsedBaseDataKey {
 public:
  explicit ParsedSetsMemberKey(const std::string *key)
      : ParsedBaseDataKey(key) {}
  explicit ParsedSetsMemberKey(const rocksdb::Slice &key)
      : ParsedBaseDataKey(key) {}
  rocksdb::Slice member() { return data_; }
};

class ParsedZSetsMemberKey : public ParsedBaseDataKey {
 public:
  explicit ParsedZSetsMemberKey(const std::string *key)
      : ParsedBaseDataKey(key) {}
  explicit ParsedZSetsMemberKey(const rocksdb::Slice &key)
      : ParsedBaseDataKey(key) {}
  rocksdb::Slice member() { return data_; }
};

typedef BaseDataKey HashesDataKey;
typedef BaseDataKey SetsMemberKey;
typedef BaseDataKey ZSetsMemberKey;